package com.ebupt.migu.music.push.service;

import com.ebupt.migu.common.push.PushContentObject;
import com.ebupt.migu.common.push.PushPolicyObject;
import com.ebupt.migu.common.push.PushRequestBody;
import com.ebupt.migu.common.util.DateUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.support.atomic.RedisAtomicInteger;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.web.util.HtmlUtils;

import javax.annotation.Resource;
import java.net.URLEncoder;
import java.util.*;

/**
 * @author tz
 */
@Slf4j
@Service
public class ProductPush {

    @Resource
    private RedisTemplate redisTemplate;
    @Autowired
    private KafkaProducerService producerService;
    @Value("${push.app_id}")
    private String PUSH_APP_ID;

    /**
     * push人群生产者
     */
    @Async("productThread")
    public void productToKafka(String taskId, Map<String, Object> strategy, List<String> splitUids,
                             List<String> splitDids, String topic) {
        // 运营频次校验
        splitUids = threshold(splitUids);
        splitDids = threshold(splitDids);
        PushContentObject pushContentObject = new PushContentObject();
        pushContent(strategy, pushContentObject);
        PushRequestBody body = new PushRequestBody();
        // policy固定值
        PushPolicyObject policy = new PushPolicyObject();
        policy.setBroadcast(false);
        policy.setIsOffline(true);
        policy.setPushNetWorkType(0);
        body.setPolicy(policy);
        body.setAppId(PUSH_APP_ID);
        body.setTaskId(taskId);
        body.setContent(pushContentObject);
        body.setUid(splitUids);
        body.setDid(splitDids);
        body.setTopic(topic);
        body.setPushChannel("1");
        body.setIsSync(true);
        if (String.valueOf(strategy.get("sendChannel")).equals("1")) {
            body.setPushAppType("music");
        }
        if (String.valueOf(strategy.get("sendChannel")).equals("2")) {
            body.setPushAppType("media");
        }
        if (String.valueOf(strategy.get("sendChannel")).equals("3")) {
            body.setPushAppType("all");
        }
        producerService.sendMessage(taskId, body);
    }

    /**
     * 构造PushContentObject实例
     *
     * @param strategy
     * @param pushContentObject
     */
    private void pushContent(Map<String, Object> strategy, PushContentObject pushContentObject) {
        // 需要进行HTML解码
        pushContentObject.setPushTitle(HtmlUtils.htmlUnescape(strategy.get("pushTitle").toString()));
        pushContentObject.setPushContent(HtmlUtils.htmlUnescape(strategy.get("pushContent").toString()));
        // 默认值是1
        pushContentObject.setPushType(1);
        // 1.前端输入链接以：mgmusic开头的，pushLinkType=100
        // 2.输入链接以http或以https开头的，pushLinkType=9
        // 20210420调整： 所有链接pushLinkType均设置为100， 以http开头的链接通过转码并拼接上固定字符串进行下发
        if (null != strategy.get("pushUrl")) {
            String pushURL = strategy.get("pushUrl").toString();
            pushContentObject.setPushLinkType(100);
            // TODO 新增mgmedia类型链接，用来下发到融媒app
            if (pushURL.startsWith("http") || pushURL.startsWith("https")) {
                try {
                    String encodeURL = URLEncoder.encode(pushURL, "UTF-8");
                    pushURL = "mgmusic://browser?url=" + encodeURL;
                } catch (Exception e) {
                    log.error("链接转码失败", e.getMessage());
                }
            }
            pushContentObject.setPushLinkAddress(pushURL);
        }

        if (null != strategy.get("picDir") && !"".equals(String.valueOf(strategy.get("picDir")))) {
            String picDir = strategy.get("picDir").toString();
            pushContentObject.setImgUrl(picDir);
        }
        pushContentObject.setPushCreateTime(System.currentTimeMillis());
    }

    /**
     * 判定当日运营是否达到阈值
     *
     * @param uids
     */
    private List<String> threshold(List<String> uids) {
        List<String> uidsTmp = new ArrayList<>();
        int limit = (null != redisTemplate.opsForValue().get("PUSH_THRESHOLD"))
                ? Integer.parseInt(String.valueOf(redisTemplate.opsForValue().get("PUSH_THRESHOLD"))) : 3;
        if (null != uids) {
            // 获取缓存的白名单集合
            Set<String> white = null;
            if (redisTemplate.hasKey("PUSH_THRESHOLD_GRAYSCALE_USERS")) {
                white = redisTemplate.opsForSet().members("PUSH_THRESHOLD_GRAYSCALE_USERS");
            }
            for (String uid : uids) {
                if (null != white && white.contains(uid)) {
                    log.info("白名单用户不受阈值管控：{}", uid);
                    continue;
                }
                // 用户的当日运营计次
                String key = "PUSH_THRESHOLD_" + DateUtil.getDateToday() + "_" + uid;
                RedisAtomicInteger redisAtomicInteger =
                        new RedisAtomicInteger(key, Objects.requireNonNull(redisTemplate.getConnectionFactory()));
                if (redisTemplate.hasKey(key) && redisAtomicInteger.get() >= limit) {
                    // 超过阈值的不下发
                    log.info("用户{}已达到当日运营次数上线，从下发名单中移除", uid);
                    uidsTmp.add(uid);
                } else {
                    // 24小时后过期
                    redisAtomicInteger.expireAt(new Date(System.currentTimeMillis() + 86400000));
                    redisAtomicInteger.getAndIncrement();
                }
            }
            uids.removeAll(uidsTmp);
        }
        return uids;
    }

}
